package com.bw.func;

import com.bw.app.bean1.DwsPopularityMetricbean;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

/**
 * 将人气指标写入Redis的Sink
 */
public class RedisSink extends RichSinkFunction<DwsPopularityMetricbean> {
    private Jedis jedis;

    @Override
    public void open(org.apache.flink.configuration.Configuration parameters) throws Exception {
        super.open(parameters);
        // 初始化Redis连接
        jedis = new Jedis("hadoop102", 6379);
    }

    @Override
    public void invoke(DwsPopularityMetricbean metric, Context context) throws Exception {
        // 构建Redis键：store:{storeId}:product:{productId}:channel:{channel}:behavior:{behaviorType}
        String key = String.format(
                "shop_id:%s:sku_id:%s:ch:%s:behavior_type:%s",
                metric.getShop_id(),
                metric.getSku_id(),
                metric.getCh(),
                metric.getBehavior_type()
        );
        
        // 存储窗口时间和计数
        jedis.hset(key, "stt", String.valueOf(metric.getStt()));
        jedis.hset(key, "edt", String.valueOf(metric.getEdt()));
        jedis.hset(key, "count", String.valueOf(metric.getCount()));
        
        // 设置过期时间（24小时）
        jedis.expire(key, 24 * 60 * 60);
        
        // 同时维护一个集合用于快速查询所有指标
        jedis.sadd("all_metrics", key);
    }

    @Override
    public void close() throws Exception {
        super.close();
        if (jedis!= null) {
            jedis.close();
        }
    }
}
